package com.carrot.mq.mqserver.datacenter;

import com.carrot.mq.common.BinaryTool;
import com.carrot.mq.common.MqException;
import com.carrot.mq.mqserver.core.MSGQueue;
import com.carrot.mq.mqserver.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.List;
import java.util.Scanner;

/**
 * @author 兴趣使然黄小黄
 * @version 1.0
 * @date 2023/8/7 2:51
 * 管理硬盘上的消息文件
 */
public class MessageFileManager {

    /**
     * 队列的统计信息
     */
    static public class Stat {
        // 总消息的数量
        public int totalCount;
        // 有效消息数量
        public int validCount;
    }

    public void init() {

    }

    /**
     * 获取到达指定队列对应的消息文件所在的路径
     */
    private String getQueueDir(String queueName) {
        return "./data/" + queueName;
    }

    /**
     * 获取该队列的数据文件路径
     */
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data.txt";
    }

    /**
     * 获取该队列的消息统计文件路径
     */
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "/queue_stat.txt";
    }

    /**
     * 读取统计文件
     */
    private Stat readStat(String queueName) {
        // 为了方便, 当前消息的统计文件用文本表示, 可以用 Scanner 读取
        Stat stat = new Stat();
        try (InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))) {
            Scanner scanner = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt();
            stat.validCount = scanner.nextInt();
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 写入统计文件
     */
    private void writeStat(String queueName, Stat stat) {
        try (OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))) {
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 创建队列的文件和目录
     */
    public void createQueueFiles(String queueName) throws IOException {
        // 创建队列对应的消息目录
        File baseDir = new File(getQueueDir(queueName));
        if (!baseDir.exists()) {
            // 不存在, 创建目录
            boolean ok = baseDir.mkdirs();
            if (!ok) {
                throw new IOException("创建目录失败! baseDir=" + baseDir.getAbsolutePath());
            }
        }
        // 创建队列数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            boolean ok = queueDataFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败! queueDataFile=" + queueDataFile.getAbsolutePath());
            }
        }
        // 创建消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            boolean ok = queueStatFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败! queueStatFile=" + queueStatFile.getAbsolutePath());
            }
        }
        // 消息统计文件设置初始值
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName, stat);
    }

    /**
     * 删除队列的文件和目录
     * 队列删除后, 对应的消息文件等被删除
     */
    public void destroyQueueFiles(String queueName) throws IOException {
        // 删除文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataFile.delete();
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatFile.delete();
        File baseDir = new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();
        if (!ok1 || !ok2 || !ok3) {
            throw new IOException("删除目录和文件夹失败! baseDir=" + baseDir.getAbsolutePath());
        }
    }

    /**
     * 检查队列的目录和文件是否存在
     */
    public boolean checkFilesExits(String queueName) {
        // 队列的数据文件和统计文件是否都存在
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            return false;
        }
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            return false;
        }
        return true;
    }

    /**
     * 把一个新的消息放到队列对应的文件中
     * @param queue     消息写入的队列
     * @param message   写的消息
     */
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        // 检查对应队列的文件是否存在
        if (!checkFilesExits(queue.getName())) {
            throw new MqException("[MessageFileManager] 队列对应的文件不存在! queueName=" + queue.getName());
        }
        // 将 Message 对象先进行序列化, 转成二进制的字节数组
        byte[] messageBinary = BinaryTool.toBytes(message);
        // 针对 queue 对象进行加锁, 如果多个线程给同一队列进行写数据, 则需要阻塞等待, 不同队列则不需要阻塞等待
        synchronized (queue) {
            // 获取当前队列数据文件的长度, 计算出该 Message 对象的 offsetBeg 和 offsetEnd
            // 新的 Message 数据写入到队列数据文件的末尾, 此时 Message 对象的 offsetBeg 就是当前文件长度 + 4
            // offsetEnd 为当前文件长度 + 4 + message 自身的长度
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            // queueDataFile.length() 获取文件的长度(字节)
            message.setOffsetBeg(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);
            // 写入消息到数据文件, 以追加的方式
            try (OutputStream outputStream = new FileOutputStream(queueDataFile, true)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    // 写入消息的长度, 占据 4 个字节
                    dataOutputStream.writeInt(messageBinary.length); // 封装了位运算的方式写入 4 个字节
                    // 写入消息本体
                    dataOutputStream.write(messageBinary);
                }
            }
            // 更新消息统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount += 1;
            stat.validCount += 1;
            writeStat(queue.getName(), stat);
        }
    }

    /**
     * 删除消息(逻辑删除, 将 isValid 设置成 0)
     * @param queue
     * @param message
     */
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException {
        synchronized (queue) {
            // 读取 queue 对应文件的数据还原 message 对象
            // 使用 RandomAccessFile 在文件的指定位置进行读写, 随机访问
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "rw")) {
                // 从文件中读取 Message 数据
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.read(bufferSrc);
                // 转化成 Message 对象
                Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);
                // 将 isValid 设置为无效
                diskMessage.setIsValid((byte) 0x0);
                // 重新写入文件
                byte[] bufferDest = BinaryTool.toBytes(diskMessage);
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.write(bufferDest);
            }
            // 更新统计文件
            Stat stat = readStat(queue.getName());
            if (stat.validCount > 0) {
                stat.validCount -= 1;
            }
            writeStat(queue.getName(), stat);
        }
    }

    /**
     * 用于在程序启动的时候, 在文件中读取所有的消息内容加载到内存中
     */
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try (InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))) {
            try (DataInputStream dataInputStream = new DataInputStream(inputStream)) {
                long currentOffset = 0L; // 记录当前文件光标
                while (true) {
                    // 读取当前消息的长度
                    // 读取到末尾的时候, 抛出 EOFException
                    int messageSize = dataInputStream.readInt();

                    // 按照长度读取消息内容
                    byte[] buffer = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffer);
                    if (messageSize != actualSize) {
                        // 检验文件格式是否错乱
                        throw new MqException("[MessageFileManager] 文件格式错误! queueName=" + queueName);
                    }
                    // 将读到的数据反序列化为 message 对象
                    Message message = (Message) BinaryTool.fromBytes(buffer);
                    // 判定是否有效
                    if (message.getIsValid() != 0x1) {
                        currentOffset += (4 + messageSize);
                        continue;
                    }
                    // 记录 offset, 与更新光标
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + messageSize);
                    currentOffset += (4 + messageSize);
                    // 将有效数据存到 list 中
                    messages.add(message);
                }
            } catch (EOFException e) {
                // 用于捕获到达文件末尾的异常
                System.out.println("[MessageFileManager] 恢复 Message 数据完成!");
            }
        }
        return messages;
    }

    /**
     * 检查当前是否要针对该队列的数据文件进行 GC
     */
    public boolean checkGC(String queueName) {
        // 根据统计文件中的有效消息数和消息总数来检验是否需要进行垃圾回收
        Stat stat = readStat(queueName);
        return stat != null && stat.totalCount > 1500 && (double) stat.validCount / (double) stat.totalCount < 0.5;
    }

    /**
     * 用于垃圾回收 GC 复制算法获取新文件的路径
     */
    private String getQueueDataNewPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }

    /**
     * 垃圾回收算法, 进行对消息数据文件的垃圾回收操作
     * 对原文件有效数据进行复制, 拷贝, 放入新文件
     * 删除旧文件, 再将新文件重命名
     * 最后, 更新消息统计文件
     */
    public void gc(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        synchronized (queue) {
            // gc 耗时比较长, 统计执行时间, 便于以后检查
            long gcBeg = System.currentTimeMillis();
            // 创建一个新的文件
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
            if (queueDataNewFile.exists()) {
                // 正常情况下, 该文件不存在, 如果存在说明上次 gc 过程中程序意外崩溃
                throw new MqException("[MessageFileManager] gc 时发现该队列的 queue_data_new 已经存在! queueName=" + queue.getName());
            }
            boolean ok = queueDataNewFile.createNewFile();
            if (!ok) {
                throw new MqException("[MessageFileManager] 创建文件失败! queueDataNewFile=" + queueDataNewFile.getAbsolutePath());
            }
            // 从旧的文件中, 读取出所有的有效消息对象
            List<Message> messages = loadAllMessageFromQueue(queue.getName());
            // 将有效消息写到新文件中
            try (OutputStream outputStream = new FileOutputStream(queueDataNewFile, true)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    for (Message message : messages) {
                        byte[] buffer = BinaryTool.toBytes(message);
                        dataOutputStream.writeInt(buffer.length); // 写入消息长度
                        dataOutputStream.write(buffer); // 写入消息内容
                    }
                }
            }
            // 删除旧的文件, 并将新的文件重命名
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            ok = queueDataOldFile.delete();
            if (!ok) {
                throw new MqException("[MessageFileManager] 删除旧数据文件失败! queueDataOldFile=" + queueDataOldFile.getAbsolutePath());
            }
            // 把 queue_data_new.txt => queue_data.txt
            ok = queueDataNewFile.renameTo(queueDataOldFile);
            if (!ok) {
                throw new MqException("[MessageFileManager] 文件重命名失败! queueDataOldFile=" + queueDataOldFile.getAbsolutePath()
                    + ", queueDataNewFile=" + queueDataNewFile.getAbsolutePath());
            }
            // 更新统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount = messages.size();
            stat.validCount = messages.size();
            writeStat(queue.getName(), stat);
            long gcEnd = System.currentTimeMillis();
            System.out.println("[MessageFileManager] gc 执行完毕! queueName=" + queue.getName() + ", time="
                + (gcEnd - gcBeg) + "ms");
        }
    }
}
